package com.jdk.learn.java8.completablefuture;

import com.jdk.learn.java8.bo.Apple;
import com.jdk.learn.java8.stream.StreamLearn;

import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static java.util.stream.Collectors.toList;

/**
 * 类功能描述:
 *
 * @author bgt
 * @version V1.0
 * @since 2019-01-15 15:34
 */
public class CompletableFutureLearn {
    private static ExecutorService executor = Executors.newFixedThreadPool(1);
    private final static Random RANDOM = new Random(System.currentTimeMillis());

    public static void main(String[] args) throws InterruptedException {
//        CompletableFutureRunAsync();
//        CompletableFutureRunAsync2();
//        CompletableFutureRunAsync3();
//        System.out.println("------------");
//        CompletableFutureRunAsync4();
//        CompletableFutureRunAsyncAndExector();
//        CompletableFutureSupplyAsync();
//        CompletableFutureSupplyAsync2();
//        CompletableFutureSupplyAsync3();
//        CompletableFutureSupplyAsync4();
//        CompletableFutureSupplyAsync5();
//        CompleteableFutureSupplyAsyncAndRunAfterBoth();
//        CompleteableFutureSupplyAsyncAndRunAfterEither();
//        CompleteableFutureSupplyAsyncAndApplyToEither();
//        CompleteableFutureSupplyAsyncAndAcceptEither();
//        CompleteableFutureAnyOf();
//        CompleteableFutureAllOf();
//        CompleteableFutureThenAccept();
//        CompleteableFutureThenApply();
        CompleteableFutureThenCombine();
        CompleteableFutureThenCompose();
    }

    /**
     * public static CompletableFuture<Void> runAsync(Runnable runnable)
     * 传入一个runnable实例对象
     * 返回CompletableFuture<Void>对象
     */
    private static void CompletableFutureRunAsync() {
        // 问题 通过main方法的形式执行 可能在主线程已经结束的情况下 还未打印出来结果
        CompletableFuture<Void> result = CompletableFuture.runAsync(() -> {
            System.out.println("执行成功");
        });
    }

    /**
     * public static CompletableFuture<Void> runAsync(Runnable runnable)
     * 传入一个runnable实例对象
     * 返回CompletableFuture<Void>对象
     */
    private static void CompletableFutureRunAsync2() throws InterruptedException {
        CompletableFuture<Void> result = CompletableFuture.runAsync(() -> {
            System.out.println("执行成功");
        });
        // 当结果未执行完毕 主线程休眠10ms
        while (!result.isDone()) {
            Thread.sleep(10);
        }
    }

    /**
     * public static CompletableFuture<Void> runAsync(Runnable runnable)
     * 传入一个runnable实例对象
     * 返回CompletableFuture<Void>对象
     */
    private static void CompletableFutureRunAsync3() throws InterruptedException {
        CompletableFuture.runAsync(() -> {
            System.out.println("执行成功");
            throw new RuntimeException("runAsync发生了异常");
        }).exceptionally(exception -> {
            System.out.println("执行失败");
            throw new RuntimeException(exception.getMessage());
        }).join();
        // join等待结果返回
    }

    private static void CompletableFutureRunAsync4() throws InterruptedException {
        CompletableFuture.runAsync(() -> {
            System.out.println("执行成功");
        }).thenAccept(System.out::println).join();
        // join等待结果返回
    }

    /**
     * public static CompletableFuture<Void> runAsync(Runnable runnable,
     * Executor executor)
     * 传入一个runnable实例对象
     * 传入一个Executor实例对象
     * 返回CompletableFuture<Void>对象
     */
    private static void CompletableFutureRunAsyncAndExector() {

        CompletableFuture.runAsync(() -> {
            System.out.println("执行成功");
        }, executor);
    }

    /**
     * public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
     * 传入一个Supplier函数
     *
     * @throws InterruptedException
     */
    private static void CompletableFutureSupplyAsync() throws InterruptedException {
        CompletableFuture.supplyAsync(() -> CompletableFutureLearn.add(StreamLearn.apples)).join();
        // join等待结果返回
    }

    /**
     * public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
     * 传入一个Supplier函数 get
     * 结果不是立即响应的
     *
     * @throws InterruptedException
     */
    private static void CompletableFutureSupplyAsync2() throws InterruptedException {
        List<CompletableFuture<String>> collect = StreamLearn.apples.stream()
                .map(apple -> CompletableFuture.supplyAsync(() -> CompletableFutureLearn.getName(apple)))
                .collect(toList());
        collect.forEach(System.out::println);
    }

    /**
     * public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
     * 传入一个Supplier函数 get
     * 返回一个CompletableFuture对象
     *
     * @throws InterruptedException
     */
    private static void CompletableFutureSupplyAsync3() throws InterruptedException {
        List<String> collect = StreamLearn.apples.stream()
                .map(apple -> CompletableFuture.supplyAsync(() -> CompletableFutureLearn.getName(apple)))
                .map(CompletableFuture::join)
                .collect(toList());

        collect.forEach(System.out::println);
    }

    /**
     * public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
     * 传入一个Supplier函数 get
     * 返回一个CompletableFuture对象
     *
     * @throws InterruptedException
     */
    private static void CompletableFutureSupplyAsync4() throws InterruptedException {
        List<String> collect = StreamLearn.apples.stream()
                .map(apple -> CompletableFuture.supplyAsync(() -> CompletableFutureLearn.getName(apple)))
                .map(future -> future.thenApply(x -> "[" + x + "]"))
                .map(CompletableFuture::join)
                .collect(toList());

        collect.forEach(System.out::println);
    }

    /**
     * public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
     * 传入一个Supplier函数 get
     * 返回一个CompletableFuture对象
     *
     * @throws InterruptedException
     */
    private static void CompletableFutureSupplyAsync5() throws InterruptedException {
        List<String> collect = StreamLearn.apples.stream()
                .map(apple -> CompletableFuture.supplyAsync(() -> CompletableFutureLearn.getName(apple)))
                .map(future -> future.thenApplyAsync(x -> "[" + x + "]"))
                .map(CompletableFuture::join)
                .collect(toList());

        collect.forEach(System.out::println);
    }

    /**
     * public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
     * <p>
     * public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
     * 接收一个Consumer函数
     * 无返回类型
     *
     * @throws InterruptedException
     */
    private static void CompletableFutureSupplyAsyncAndThenAccept() throws InterruptedException {
        CompletableFuture.supplyAsync(() -> CompletableFutureLearn.add(StreamLearn.apples))
                .thenAccept(System.out::println).join();
        // join等待结果返回
    }

    /**
     * 在两者都执行完后执行 Action
     * public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other,
     * Runnable action) {
     * 传入 一个CompletionStage对象 other
     * 一个Runnable类型的Action
     * <p>
     * CompletableFuture.supplyAsync(A).runafterBoth(other,action)
     * <p>
     * 执行顺序
     * A->other->action
     */
    private static void CompleteableFutureSupplyAsyncAndRunAfterBoth() {
        CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " is running. 111");
            imitateTime(100l);
            return 1;
        }).runAfterBoth(CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + " is running. 222");
            return 2;
        }), () -> System.out.println("done")).join();
    }

    /**
     * 在两者有一个执行完后执行 Action
     * public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,
     * Runnable action) {
     * 传入 一个CompletionStage对象 other
     * 一个Runnable类型的Action
     * <p>
     * CompletableFuture.supplyAsync(A).runafterBoth(other,action)
     * <p>
     * 执行顺序
     * A->action
     * 或
     * other->action
     */
    private static void CompleteableFutureSupplyAsyncAndRunAfterEither() {
        CompletableFuture.supplyAsync(() -> {
            imitateTime(103l);
            System.out.println(Thread.currentThread().getName() + " is running. 111  " + LocalDateTime.now());
            return 1;
        }).runAfterEither(CompletableFuture.supplyAsync(() -> {
            imitateTime(202l);
            System.out.println(Thread.currentThread().getName() + " is running. 222  " + LocalDateTime.now());
            return 2;
        }), () -> System.out.println("done  " + LocalDateTime.now())).join();
    }


    /**
     * 获取两个中的一个 (目前是那个快 取那个),之后调用fn函数
     * public <U> CompletableFuture<U> applyToEither(
     * CompletionStage<? extends T> other, Function<? super T, U> fn)
     * 传入 一个CompletionStage对象 other
     * 一个Function类型的fn
     * <p>
     * 执行行为：
     * CompletableFuture.supplyAsync(A).applyToEither(other,fn)
     * AB那个先执行完，用那个
     * A 先 则fn=0
     * B 先 则fn=200
     */
    private static void CompleteableFutureSupplyAsyncAndApplyToEither() {
        CompletableFuture.supplyAsync(() -> {
            Integer result = 0;
            imitateTime(100l);
            System.out.println("i am 0");
            return result;
        }).applyToEither(CompletableFuture.supplyAsync(() -> {
            imitateTime(101l);
            System.out.println("i am 2");
            return 2;
        }), (x) -> {
            return x * 100;
        }).thenAccept(System.out::println).join();
    }

    /**
     * public CompletableFuture<Void> acceptEither(
     * CompletionStage<? extends T> other, Consumer<? super T> action)
     * 传入 CompletionStage类型的 other
     * Consumer类型的 action
     * <p>
     * CompletableFuture.supplyAsync(A).acceptEither(other,action)
     * A 如先执行完 action 根据A的结果执行
     * other如果先执行完 action根据other的结果执行
     */
    private static void CompleteableFutureSupplyAsyncAndAcceptEither() {
        CompletableFuture.supplyAsync(() -> {
            Integer result = 0;
            imitateTime(102l);
            System.out.println("i am 0");
            return result;
        }).acceptEither(CompletableFuture.supplyAsync(() -> {
            imitateTime(101l);
            System.out.println("i am 2");
            return 2;
        }), (result) -> {
            System.out.println("result:" + result);
        }).join();
    }

    /**
     * 任何一个执行完thenrun
     */
    private static void CompleteableFutureAnyOf() {
        List<CompletableFuture<Float>> collect = Arrays.asList(1f, 2f, 3f, 4f)
                .stream()
                .map(i -> CompletableFuture.supplyAsync(CompletableFutureLearn::imitateTime2))
                .collect(toList());

        CompletableFuture.anyOf(collect.toArray(new CompletableFuture[collect.size()]))
                .thenRun(() -> System.out.println("done")).join();

    }

    /**
     *  所有都执行完 thenrun
     */
    private static void CompleteableFutureAllOf() {
        List<CompletableFuture<Float>> collect = Arrays.asList(1f, 2f, 3f, 4f)
                .stream()
                .map(i -> CompletableFuture.supplyAsync(CompletableFutureLearn::imitateTime2))
                .collect(toList());

        CompletableFuture.allOf(collect.toArray(new CompletableFuture[collect.size()]))
                .thenRun(() -> System.out.println("done")).join();

    }
    /**
     * public CompletableFuture<Void> thenAccept(Consumer<? super T> action)
     *  传入一个Consumer对象
     */
    private static void CompleteableFutureThenAccept() {
        CompletableFuture.supplyAsync(()->{
            return RANDOM.nextFloat();
        }).thenAccept(System.out::println).join();
    }

    /**
     *  public <U> CompletableFuture<U> thenApply(
     *         Function<? super T,? extends U> fn)
     *  传入一个Function对象
     */
    private static void CompleteableFutureThenApply() {
        CompletableFuture.supplyAsync(()->{
            return RANDOM.nextFloat();
        }).thenApply(String::valueOf).join();
    }

    /**
     *  public <U,V> CompletableFuture<V> thenCombine(
     *         CompletionStage<? extends U> other,
     *         BiFunction<? super T,? super U,? extends V> fn)
     *  传入一个CompletionStage对象
     *  传入一个BiFunction对象
     */
    private static void CompleteableFutureThenCombine() {
        CompletableFuture.supplyAsync(()->{
            return 10f;
        }).thenCombine(CompletableFuture.supplyAsync(()->{
            return 100f;
        }),(x,y)->{
            return x*y;
        }).thenAccept(System.out::println).join();
    }
    /**
     *
     *  public <U> CompletableFuture<U> thenCompose(
     *         Function<? super T, ? extends CompletionStage<U>> fn)
     *  传入一个Function对象
     */
    private static void CompleteableFutureThenCompose() {
        CompletableFuture.supplyAsync(()->{
            return 10f;
        }).thenCompose((x)->{
           return CompletableFuture.supplyAsync(()->{
               return x*100;
           });
        }).thenAccept(System.out::println).join();
    }

    public static List<String> add(List<Apple> appleList) {
        return appleList.stream().map(Apple::getName).collect(toList());
    }

    public static String getName(Apple apple) {
        System.out.println("getName");
        return apple.getName();
    }

    public static Apple plusWeight(Apple apple) {
        apple.setWeight(apple.getWeight() * 10l);
        return apple;
    }

    private static void imitateTime(Long time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public static void print(Object x) {
        System.out.println("done:"+x);
    }

    private static Float imitateTime2() {
        try {
            Thread.sleep(100);
            Float result = RANDOM.nextFloat();
            System.out.println(result);
            return result;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 0f;
    }
}
